package com.leyantech.ecp.agenx.translator.handler;

import static com.leyantech.ecp.agenx.common.config.ConfigManager.DIALOGUE_TOPIC;

import com.leyantech.ai.DialogueQuestion;
import com.leyantech.chaos.core.util.StatsdTool;
import com.leyantech.chaos.kafka.KafkaDialogueProducerClient;
import com.leyantech.ecp.agenx.common.converter.input.DialogueQuestionConverter;
import com.leyantech.ecp.agenx.common.handler.ActionHandler;
import com.leyantech.ecp.agenx.common.message.EnumWsFnType;
import com.leyantech.ecp.agenx.common.utils.PbPrintHelper;
import com.leyantech.utility.constructs.Constructs;

import com.google.inject.Inject;
import com.google.protobuf.Any;
import io.vertx.core.json.JsonObject;
import lombok.extern.slf4j.Slf4j;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-08.
 */
@Slf4j
public class BuyerMessageHandler extends ActionHandler {

  private final DialogueQuestionConverter dialogueQuestionConverter;

  private final KafkaDialogueProducerClient dialogueProducerClient;

  @Inject
  public BuyerMessageHandler(
      DialogueQuestionConverter dialogueQuestionConverter,
      KafkaDialogueProducerClient dialogueProducerClient) {
    this.dialogueQuestionConverter = dialogueQuestionConverter;
    this.dialogueProducerClient = dialogueProducerClient;
  }

  @Override
  public EnumWsFnType actionType() {
    return EnumWsFnType.NORMAL_BUYER_MSG;
  }

  @Override
  public void handle(JsonObject jsonObject) {

    final DialogueQuestion dialogueQuestion = dialogueQuestionConverter.apply(jsonObject);

    if (dialogueQuestion == DialogueQuestion.getDefaultInstance()) {
      return;
    }

    final String key = dialogueQuestion.getSessionId();
    dialogueProducerClient.send(DIALOGUE_TOPIC.get(), key, Any.pack(dialogueQuestion), null,
        (metadata, throwable) -> {
          if (throwable != null) {
            StatsdTool.increment(Constructs.LAIN_PROCNAME, "message", dialogueQuestion.getServiceType().name(), "failed");
            log.error("send to dialogue failed", throwable);
          } else {
            StatsdTool.increment(Constructs.LAIN_PROCNAME, "message", dialogueQuestion.getServiceType().name(), "succeed");
            log.info("send to dialogue succeed, {}", PbPrintHelper.lazyToJson(dialogueQuestion));
          }
        }
    );
  }

}
